package com.iwhalecloud.uncc.mq.consumer;

import com.alibaba.druid.util.lang.Consumer;
import com.iwhalecloud.uncc.common.SysConstant;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import com.iwhalecloud.uncc.model.MqClusterConfig;
import com.iwhalecloud.uncc.model.MqConsumerConfig;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

@Slf4j
public class RocketMqConsumer implements MqConsumer {

	private DefaultMQPushConsumer consumer; // 默认push消费者
	private MqClusterConfig clusterConfig;
	private MqConsumerConfig consumerConfig; // 配置
	private MessageListener messageListener; // 监听器

	public RocketMqConsumer(MqClusterConfig mqClusterConfig, MqConsumerConfig consumerConfig,
			MessageListener messageListener) {
		this.clusterConfig = mqClusterConfig;
		this.consumerConfig = consumerConfig;
		this.messageListener = messageListener;
	}

	@Override
	public void init() {
		log.info("init consumer start, consumerName: " + consumerConfig.getConsumerName());
		try {
			if (consumer == null) {
				consumer = new DefaultMQPushConsumer();
				consumer.setConsumerGroup(consumerConfig.getConsumerName());
				consumer.setNamesrvAddr(clusterConfig.getNameserv());
				consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
				consumer.setConsumeMessageBatchMaxSize(consumerConfig.getConsumeBatchSize());
				consumer.setPullBatchSize(consumerConfig.getConsumeBatchSize());
				consumer.setMessageModel(SysConstant.MQ_MESSAGE_MODEL_BROADCASTING.equalsIgnoreCase(consumerConfig.getMessageModel())?
						MessageModel.BROADCASTING : MessageModel.CLUSTERING);
				consumer.setConsumeTimeoutInSec(7200);
				consumer.setConsumeThreadMax(consumerConfig.getConsumeThreadMax());
				consumer.setConsumeThreadMin(consumerConfig.getConsumeThreadMin());
				consumer.subscribe(consumerConfig.getTopic(), consumerConfig.getTag());
			}
			log.info("init consumer success, consumerName: " + consumerConfig.getConsumerName());
		} catch (Exception e) {
			log.info("init consumer exception , consumerName: " + consumerConfig.getConsumerName(), e);
		}
	}

	@Override
	public void listen() throws Exception {
		consumer.registerMessageListener(messageListener);
		consumer.start();
	}

	@Override
	public void shutdown() throws Exception {
		if (consumer != null) {
			try {
				consumer.shutdown();
			} catch (Exception e) {
				log.error(ExceptionUtils.getStackTrace(e));
			}
		}
	}
}
